package com.spzx.common.rabbitmq;

import com.spzx.common.rabbitmq.domain.SpzxCorrelationData;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class SpzxRabbitmqService {
    @Resource
    RabbitTemplate rabbitTemplate;
    @Resource
    RedisTemplate redisTemplate;
    //当前业务类初始化后 配置mq的生产者确认
    @PostConstruct
    public void init(){
        //必须在配置文件中开启生产者回调
        //消息是否达到交换机的回调
        rabbitTemplate.setConfirmCallback((correlationData, ack,cause)->{
            //消息到达交换机：直接return
            if(ack){
                return;
            }

            retry(correlationData.getId(), cause , "交换机");

//                log.error("消息未到达交换机，原因："+cause);
//                System.out.println("correlationData: "+ correlationData + "， ack： "+ack);

        });
        //消息未到队列的回调
        rabbitTemplate.setReturnsCallback((returned)->{
            //消息未到达队列：尝试重新发送消息(重试次数不能超过3次)
//            log.error("消息未到达队列，原因："+returned.getReplyText() +" , replyCode: "+ returned.getReplyCode()+
//                    " , exchange: "+ returned.getExchange() + " , routingKey: "+ returned.getRoutingKey()+" , message: "+
//                    returned.getMessage());

           retry(returned.getMessage().getMessageProperties()
                   .getHeader("spring_returned_message_correlation")
                   .toString() , "应答码:"+returned.getReplyCode()+",应答内容:"+returned.getReplyText() , "队列");
        });

    }

    private void retry(String correlationId , String cause , String title) {

        Object obj = redisTemplate.opsForValue().get("spzx:mq:retry:"+correlationId);
        if(obj!=null){
            SpzxCorrelationData spzxCorrelationData = (SpzxCorrelationData) obj;
            //不需要重试
            if(!spzxCorrelationData.isRetry()){
                log.error("消息未到达{}，原因:{} , 不重试",title,cause);
                return;
            }
            //判断重试次数：
            //超过允许的重试次数
            if(spzxCorrelationData.getRetryCount()>=spzxCorrelationData.getRetryLimit()){
                log.error("消息未到达{}，原因："+ cause +" , 重试次数 {} 超过允许次数 {}" ,title,spzxCorrelationData.getRetryCount(),
                        spzxCorrelationData.getRetryLimit());
                return;
            }
            //重试：
            spzxCorrelationData.setRetryCount(spzxCorrelationData.getRetryCount()+1);
            //设置到redis中 更新spzxCorrelationData的重试次数
            redisTemplate.opsForValue().set("spzx:mq:retry:"+correlationId ,spzxCorrelationData );
            sendMsg(spzxCorrelationData);
        }


    }

    public void sendMsg(String exchange , String routingKey , Object message){
        rabbitTemplate.convertAndSend(exchange ,routingKey ,message);
    }
    //CorrelationData: rabbitmq发送消息时，如果传入了CorrelationData，消息发送失败时，会自动将该对象传给回调方法
    //   通过CorrelationData携带的参数 可以尝试重新发送消息
    public void sendMsg(SpzxCorrelationData correlationData){
        //    //可以使用SpzxCorrelationData的id值 将SpzxCorrelationData对象缓存  4b8cc5f3-9732-45fa-bcc5-95661be39c93
        //        //spring_returned_message_correlation -> 4b8cc5f3-9732-45fa-bcc5-95661be39c93
        //        // 生产者未到达队列的回调，只有 SpzxCorrelationData的id 没有SpzxCorrelationData对象
        //发送消息前  使用correlationData的id 将 correlationData  映射 缓存(消息未到达队列回调中需要获取correlationData使用)
        redisTemplate.opsForValue().set("spzx:mq:retry:"+correlationData.getId() ,correlationData ,
                10 , TimeUnit.MINUTES);
        rabbitTemplate.convertAndSend(correlationData.getExchange() ,correlationData.getRoutingKey() ,
                correlationData.getMessage(),correlationData);
    }
}
